package org.example.kafkastudy.consumer;

import com.alibaba.fastjson.JSON;
import com.sun.org.apache.xpath.internal.operations.Or;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.example.kafkastudy.model.Order;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class Consumer {

    @KafkaListener(topics = {"test-sdk-send-topic"}, groupId = "order-service")
//    @KafkaListener(topics = {"${kafka.topic.order.create}}"}, groupId = "order-service")  // 支持通配符，从 application.yaml 中读取数据
    public void listen(String msg) {
        System.out.println("消费到消息：" + msg);
    }


    @KafkaListener(topics = {"test-consume-params"}, groupId = "order-service")
    public void listen2(ConsumerRecord<String, String> record,
                        @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                        @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key,
                        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) String partition,
                        @Payload String msg) {
        System.out.println("消费到消息，record：" + record);
        System.out.println("消费到消息，topic：" + topic);
        System.out.println("消费到消息，partition：" + partition);
        System.out.println("消费到消息，key：" + key);
        System.out.println("消费到消息，msg：" + msg);
    }

//    @KafkaListener(topics = {"test-consume-manual-ack"}, groupId = "order-service")
    public void listen3(String msg, Acknowledgment ack) throws InterruptedException {
        System.out.println("消费到消息：" + msg);

        // 模拟业务逻辑
        Thread.sleep(1000);
        int result = 100 / Integer.parseInt(msg);
        System.out.println("业务逻辑处理结果：" + result);

        ack.acknowledge(); // 业务逻辑处理成功后，再返回 ack
    }


//    @KafkaListener(groupId = "order.service",
//            topicPartitions = {
//                    @TopicPartition(
//                            topic = "test-set-offset",
//                            partitions = {"0", "1"}, // 没有设置 initialOffset 的 Partition 将不会读取历史消息
//                            partitionOffsets = {
//                                    @PartitionOffset(partition = "2", initialOffset = "3"), // 从指定 offset 开始读取，哪怕是已经消费过的
//                                    @PartitionOffset(partition = "3", initialOffset = "3")
//                            })
//            })
    public void listen4(ConsumerRecord<String, String> record) {
        System.out.println("test-set-offset，收到消息：" + record.value() + "，Partition：" + record.partition());
    }

    @KafkaListener(topics = {"test-batch-consume"}, groupId = "order-service")
    public void listen5(List<ConsumerRecord<String, String>> records) {
        System.out.println("批量消费，records.size() = " + records.size() + "，records = " + records);
    }


//    @KafkaListener(topics = {"order.create"}, groupId = "stock-service", containerFactory = "orderKafkaListenerContainerFactory")
//    public void listenOrderCreate(ConsumerRecord<String, String> record) {
//        System.out.println("消费 order.create 消息，record：" + record);
//    }
//
//    @KafkaListener(topics = {"pay.success"}, groupId = "stock-service", containerFactory = "payKafkaListenerContainerFactory")
//    public void listenPaySuccess(ConsumerRecord<String, String> record) {
//        System.out.println("消费 pay.success 消息，record：" + record);
//    }


    @KafkaListener(topics = {"order.create"}, groupId = "pay-service")
    @SendTo("pay.success")
    public String listenOrderCreate(ConsumerRecord<String, String> record) throws InterruptedException {
        System.out.println("支付服务，消费 order.create 消息，record：" + record.value());

        // 模拟支付逻辑
        Order order = JSON.parseObject(record.value(), Order.class);
        if (!"init".equals(order.getStatus())) {
            return record.value();
        }
        Thread.sleep(200);
        System.out.println("支付服务，订单（" + order.getId() + "）支付完成。");
        order.setStatus("paid");

        return JSON.toJSONString(order);
    }

    @KafkaListener(topics = {"pay.success"}, groupId = "stock-service")
    public void listenPaySuccess(ConsumerRecord<String, String> record) throws InterruptedException {
        System.out.println("库存服务，消费 pay.success 消息，record：" + record.value());

        Order order = JSON.parseObject(record.value(), Order.class);
        if (!"paid".equals(order.getStatus())) {
            return;
        }
        Thread.sleep(200);
        System.out.println("库存服务，扣减库存成功，商品：" + order.getProductId() + "，数量：" + order.getNum());
    }
}
